package com.ktjy.lotrabbitmqconsumer7005.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ktjy.lotrabbitmqconsumer7005.pojo.LogDTO;
import com.ktjy.lotrabbitmqconsumer7005.pojo.MqttMessage;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

@Service
class EMQXConsumerService {

    private final ObjectMapper mapper = new ObjectMapper();

    /**
     * 队列中模拟的报文
     * {
     *   "level": 2,
     *   "content":"aaa"
     * }
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = "emqx_temp_queue")
    public void processMessage(Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String json = new String(message.getBody(), StandardCharsets.UTF_8);
        System.out.println("原始JSON: " + json);
        System.out.println("消息属性: " + message.getMessageProperties());
        ObjectMapper mapper = new ObjectMapper();
        // 解析外层消息
        MqttMessage mqttMessage = mapper.readValue(json, MqttMessage.class);
        // 解析内层负载
        LogDTO payload = mapper.readValue(mqttMessage.getPayload(), LogDTO.class);
        System.out.println("用户名: " + mqttMessage.getUsername());
        System.out.println("主题: " + mqttMessage.getTopic());
        System.out.println("消息级别: " + payload.getLevel());
        System.out.println("消息内容: " + payload.getContent());
        //执行消息的业务逻辑，可以是增加
        if (!payload.getLevel().equals("1")) {
            channel.basicAck(deliveryTag, false);
            System.out.println("消息已确认");
        } else {
            channel.basicNack(deliveryTag, false, true);
            System.out.println("消息处理失败，已拒绝并重新入队");
        }
        try {
            Thread.sleep(4000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}